package com.gjy.design.observer.disruptor;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-11-08 09:16:37
 */
public class TopicManager {

    public final ConcurrentMap<String, CopyOnWriteArraySet<Subscriber>> topicSubscribers = new ConcurrentHashMap<>();

    public void subscribe(String topic, Subscriber subscriber) {
        topicSubscribers.compute(topic, (k, v) -> {
            if (v == null) {
                v = new CopyOnWriteArraySet<>();
            }
            v.add(subscriber);
            return v;
        });
    }

    public void unsubscribe(String topic, Subscriber subscriber) {
        topicSubscribers.computeIfPresent(topic, (k, v) -> {
            v.remove(subscriber);
            return v.isEmpty() ? null : v;
        });
    }

    public void unsubscribeAll(Subscriber subscriber) {
        for (String topic : topicSubscribers.keySet()) {
            unsubscribe(topic, subscriber);
        }
    }

    public CopyOnWriteArraySet<Subscriber> getSubscribers(String topic) {
        return topicSubscribers.get(topic);
    }

    public boolean hasSubscribers(String topic) {
        CopyOnWriteArraySet<Subscriber> subscribers = topicSubscribers.get(topic);
        return subscribers != null && !subscribers.isEmpty();
    }

}
